package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.function.OverSpeedRichFilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import utils.KafkaUtil;

public class OverSpeedController {

    public static void main(String[] args) throws  Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        //2.读取数据源
        DataStream<String> trafficStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");
//        trafficStream.print();
        //3.数据处理
        DataStream<String> resultStream = handle(trafficStream);
        resultStream.print();
        //4.数据输出

        //5.执行
        env.execute("OverSpeedController");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        // 封装成实体类
        SingleOutputStreamOperator<MonitorInfo> mapStream = stream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String value) throws Exception {
                return JSON.parseObject(value, MonitorInfo.class);
            }
        });

        //2.过滤
        SingleOutputStreamOperator<MonitorInfo> filterStream = mapStream.filter(new OverSpeedRichFilterFunction<>());

        SingleOutputStreamOperator<String> map = filterStream.map(new MapFunction<MonitorInfo, String>() {
            @Override
            public String map(MonitorInfo value) throws Exception {
                System.out.println(value);
                return null;
            }
        });

        return map;
    }
}
